package com.wugui.datax.admin.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.api.R;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.wugui.datatx.core.biz.model.ReturnT;
import com.wugui.datax.admin.core.util.I18nUtil;
import com.wugui.datax.admin.entity.JobInfo;
import com.wugui.datax.admin.mapper.JobDatasourceMapper;
import com.wugui.datax.admin.entity.JobDatasource;
import com.wugui.datax.admin.mapper.JobInfoMapper;
import com.wugui.datax.admin.service.JobDatasourceService;
import com.wugui.datax.admin.tool.query.BaseQueryTool;
import com.wugui.datax.admin.tool.query.HBaseQueryTool;
import com.wugui.datax.admin.tool.query.MongoDBQueryTool;
import com.wugui.datax.admin.tool.query.QueryToolFactory;
import com.wugui.datax.admin.util.AESUtil;
import com.wugui.datax.admin.util.JdbcConstants;

import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;

import java.io.IOException;
import java.util.List;

/**
 * Created by jingwk on 2020/01/30
 */
@Service
@Transactional
public class JobDatasourceServiceImpl extends ServiceImpl<JobDatasourceMapper, JobDatasource> implements JobDatasourceService {

    @Resource
    private JobDatasourceMapper datasourceMapper;

    @Resource
    private JobInfoMapper jobInfoMapper;

    @Override
    public Boolean dataSourceTest(JobDatasource jobDatasource) throws IOException {
        if (JdbcConstants.HBASE.equals(jobDatasource.getDatasource())) {
            return new HBaseQueryTool(jobDatasource).dataSourceTest();
        }
        String userName = AESUtil.decrypt(jobDatasource.getJdbcUsername());
        //  判断账密是否为密文
        if (userName == null) {
            jobDatasource.setJdbcUsername(AESUtil.encrypt(jobDatasource.getJdbcUsername()));
        }
        String pwd = AESUtil.decrypt(jobDatasource.getJdbcPassword());
        if (pwd == null) {
            jobDatasource.setJdbcPassword(AESUtil.encrypt(jobDatasource.getJdbcPassword()));
        }
        if (JdbcConstants.MONGODB.equals(jobDatasource.getDatasource())) {
            return new MongoDBQueryTool(jobDatasource).dataSourceTest(jobDatasource.getDatabaseName());
        }
        BaseQueryTool queryTool = QueryToolFactory.getByDbType(jobDatasource);
        return queryTool.dataSourceTest();
    }

    @Override
    public R<Boolean> update(JobDatasource datasource) {
        if (StringUtils.isBlank(datasource.getJdbcUsername())) {
            return R.failed(I18nUtil.getString("system_please_input") + I18nUtil.getString("datasource_field_jdbc_username"));
        }
        if (StringUtils.isBlank(datasource.getJdbcPassword())) {
            return R.failed(I18nUtil.getString("system_please_input") + I18nUtil.getString("datasource_field_jdbc_password"));
        }
        if (StringUtils.isBlank(datasource.getJdbcUrl())) {
            return R.failed(I18nUtil.getString("system_please_input") + I18nUtil.getString("datasource_field_jdbc_url"));
        }

        // 由于账号/密码在写入数据库前会自动加密，未修改的情况下不要更新账号/密码，否则会造成二次加密
        JobDatasource d = datasourceMapper.selectById(datasource.getId());
        if (datasource.getJdbcUsername().equals(d.getJdbcUsername())) {
            datasource.setJdbcUsername(null);
        }
        if (datasource.getJdbcPassword().equals(d.getJdbcPassword())) {
            datasource.setJdbcPassword(null);
        }
        List<JobInfo> jobInfos1 = jobInfoMapper.listByReaderDatasourceId(datasource.getId());
        jobInfos1.forEach(jobInfo -> updateJobInfoReader(jobInfo, datasource));
        List<JobInfo> jobInfos2 = jobInfoMapper.listByWriterDatasourceId(datasource.getId());
        jobInfos2.forEach(jobInfo -> updateJobInfoWriter(jobInfo, datasource));
        return R.ok(updateById(datasource));
    }

    private void updateJobInfoReader(JobInfo jobInfo, JobDatasource datasource) {
        // 更新任务job_json内的reader
        String jobJson = jobInfo.getJobJson();
        JSONObject json = JSONObject.parseObject(jobJson);
        JSONObject job = json.getJSONObject("job");
        JSONArray contents = job.getJSONArray("content");
        for (int i = 0; i < contents.size(); i++) {
            JSONObject content = contents.getJSONObject(i);
            JSONObject reader = content.getJSONObject("reader");
            JSONObject params = JSONObject.parseObject(reader.getString("parameter"));
            if (StringUtils.isNotBlank(datasource.getJdbcUsername())) {
                params.put("username", AESUtil.encrypt(datasource.getJdbcUsername()));
            }
            if (StringUtils.isNotBlank(datasource.getJdbcPassword())) {
                params.put("password", AESUtil.encrypt(datasource.getJdbcPassword()));
            }
            JSONArray connections = params.getJSONArray("connection");
            for (int j = 0; j < connections.size(); j++) {
                JSONObject connection = connections.getJSONObject(j);
                JSONArray jdbcUrls = connection.getJSONArray("jdbcUrl");
                for (int k = 0; k < jdbcUrls.size(); k++) {
                    jdbcUrls.set(i, datasource.getJdbcUrl());
                }
            }
            reader.put("parameter", params);
            content.put("reader", reader);
        }
        job.put("content", contents);
        json.put("job", job);
        jobInfo.setJobJson(json.toJSONString());
        jobInfoMapper.update(jobInfo);
    }

    private void updateJobInfoWriter(JobInfo jobInfo, JobDatasource datasource) {
        // 更新任务job_json内的writer
        String jobJson = jobInfo.getJobJson();
        JSONObject json = JSONObject.parseObject(jobJson);
        JSONObject job = json.getJSONObject("job");
        JSONArray contents = job.getJSONArray("content");
        for (int i = 0; i < contents.size(); i++) {
            JSONObject content = contents.getJSONObject(i);
            JSONObject writer = content.getJSONObject("writer");
            JSONObject params = JSONObject.parseObject(writer.getString("parameter"));
            if (StringUtils.isNotBlank(datasource.getJdbcUsername())) {
                params.put("username", AESUtil.encrypt(datasource.getJdbcUsername()));
            }
            if (StringUtils.isNotBlank(datasource.getJdbcPassword())) {
                params.put("password", AESUtil.encrypt(datasource.getJdbcPassword()));
            }
            JSONArray connections = params.getJSONArray("connection");
            for (int j = 0; j < connections.size(); j++) {
                JSONObject connection = connections.getJSONObject(j);
                connection.put("jdbcUrl", datasource.getJdbcUrl());
            }
            writer.put("parameter", params);
            content.put("writer", writer);
        }
        job.put("content", contents);
        json.put("job", job);
        jobInfo.setJobJson(json.toJSONString());
        jobInfoMapper.update(jobInfo);
    }

    @Override
    public List<JobDatasource> selectAllDatasource() {
        return datasourceMapper.selectList(null);
    }

}